#include <vector>
#include <mutex>
#include <thread>
#include <chrono>
#include <random>
#include <functional>
#include <atomic>
#include <iostream>
#include <sys/epoll.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <fcntl.h>
#include <errno.h>


namespace net {
    struct NetWordRequest{
        int fd;
        std::string data;
        std::chrono::steady_clock::time_point timeout;

        NetWordRequest(int fd, std::string data, std::chrono::steady_clock::time_point timeout) : 
            fd(fd), data(data), timeout(timeout) {}
    };

class WorkStealingQue{
private:
    mutable std::mutex mutex_;
    std::deque<std::unique_ptr<NetWordRequest>> queue_;
public:
    void push(NetWordRequest request) {
        std::lock_guard<std::mutex> lock(mutex_);
        queue_.push_front(request);
    }

    std::unique_ptr<NetWordRequest> pop() {
        std::lock_guard<std::mutex> lock(mutex_);
        if (queue_.empty()) {
            return nullptr;
        }
        std::unique_ptr<NetWordRequest> req = std::move(queue_.front());
        queue_.pop_front();
        return req;
    }
    std::unique_ptr<NetWordRequest> steal() {
        std::lock_guard<std::mutex> lock(mutex_);
        if (queue_.empty()) return nullptr;
        auto req = std::move(queue_.back());
        queue_.pop_back();
        return req;
    }

    bool empty() const{
        std::lock_guard<std::mutex> lock(mutex_);
        return queue_.empty();
    }
    size_t size() const{
        std::lock_guard<std::mutex> lock(mutex_);
        return queue_.size();
    }
};
template<typename T, size_t MAX_SIZE = 1024>
class LockFreeWorkStealingQueue {
private:
    static_assert(MAX_SIZE > 0, "MAX_SIZE must be greater than 0");
    struct alignas(64) Slot {
        std::atomic<T*> data{nullptr};
    };
    Slot slots[MAX_SIZE];
    std::atomic<size_t> head{0};
    std::atomic<size_t> tail{0};
    static constexpr size_t mask = MAX_SIZE - 1;
public:
    LockFreeWorkStealingQueue() = default;
    ~LockFreeWorkStealingQueue() {
        T* item;
        while ((item = pop()) != nullptr) {
            delete item;
        }
    }
    bool push(T* item) {
        size_t head = head.load(std::memory_order_relaxed);
        size_t tail = tail.load(std::memory_order_acquire);

        if (head - tail >= MAX_SIZE) return false;

        slots[head & mask].data.store(item, std::memory_order_relaxed);
        head.store(head + 1, std::memory_order_release);
        return true;
    }
    T* pop() {
        size_t head = head.load(std::memory_order_relaxed);
        size_t tail = tail.load(std::memory_order_acquire);

        if (head == tail) 
        {
            return nullptr;
        }
        head.store(head - 1, std::memory_order_relaxed);
        T* item = slots[head & mask].data.load(std::memory_order_acquire);
        return item;
    }
    T* steal() {
        size_t head = head.load(std::memory_order_relaxed)
        size_t taile = tail.load(std::memory_order_acquire);
        if (tail >= head) return nullptr;
        T* item = slots[tail & mask].data.load(std::memory_order_acquire);
        if (tail.compare_exchange_strong(tail, tail + 1, std::memory_order_acquire, std::memory_order_relaxed))
        {
            return item;
        }
        return nullptr;
    }
    bool empty() const{
        return head.load(std::memory_order_relaxed) == tail.load(std::memory_order_acquire);
    }
    size_t size() const{
        return head.load(std::memory_order_relaxed) - tail.load(std::memory_order_acquire);
    }
};

class LockFreeWorkStealingQueueDynamic {
private:
    struct Node {

    };
public:

};
class WorkStealingThreadPool {
private:
    std::vector<std::thread> threads_;
    std::vector<std::unique_ptr<LockFreeWorkStealingQueue>> queues_;
    std::atomic<bool> running_;
    std::atomic<int> next_queue_;
    std::mutex mutex_;
    std::condition_variable cv_;

    using RequestHandler = std::function<void(std::unique_ptr<NetWordRequest>)>;
    RequestHandler handler_;

    std::atomic<uint64_t> total_processed_;
    std::atomic<uint64_t> total_stolen_;
public:
    WorkStealingThreadPool(size_t num_threads, RequestHandler handler) 
        : handler_(std::move(handler)){
            queues_.reserve(num_threads);
            threads_.reserve(num_threads);
            for (size_t i = 0; i < num_threads; ++i) {
                queues_.emplace_back(std::make_unique<WorkStealingQue>());
            }
            cv_ = std::condition_variable();
        }

    ~WorkStealingThreadPool() {
        running_.store(false, std::memory_order_relaxed);
        for (auto& thread : threads_) {
            if (thread.joinable()) thread.join();
        }
    }
    void start() {
        running_.store(true, std::memory_order_relaxed);
        for (size_t i = 0; i < queues_.size(); ++i) {
            threads_.emplace_back(&WorkStealingThreadPool::workerThread, this, i);
        }
    }
    void stop() {
        running_.store(false, std::memory_order_relaxed);
        for (auto& thread : threads_) {
            if (thread.joinable()) thread.join();
        }
        threads_.clear();
    }
    void submit(std::unique_ptr<NetWordRequest> request) {
        int idx = next_queue_.fetch_add(1) % queues_.size();
        std::lock_guard<std::mutex> lock(mutex_);
        queues_[idx]->push(std::move(request));
        cv_.notify_one();
    }
private:
    void workerThread(size_t idx) {
        std::random_device rd;
        std::mt19937 gen(rd());

        while (running_) {
            // 1. 从自己队列中取
            std::unique_ptr<NetWordRequest> req;
            req = queues_[idx]->pop();
            
            // 2. 从其他队列中偷
            
            if (!req && queues_.size() > 1) {
                std::uniform_int_distribution<> dis(0, queues_.size() - 1);
                for (size_t attempts = 0; attempts < queues_.size(); ++attempts) {
                    size_t target_que;
                    do {
                        target_que = dis(gen);
                    } while (target_que == idx);
                    req = queues_[target_que]->steal();
                    if (req) {
                        total_stolen_.fetch_add(1);
                        break;
                    }
                }
            }
            // 3. 如果都没有，则等待
            if (!req) {
                std::unique_lock<std::mutex> lock(mutex_);
                cv_.wait(lock, [this, idx]{
                    return !running_ || !queues_[idx]->empty();
                });
            }
            else {
                // 4. 处理请求
                try {
                    handler_(std::move(req));
                    // 5. 统计处理量
                    total_processed_.fetch_add(1);
                } catch (const std::exception& e) {
                    std::cerr << "Worker thread " << idx << " caught exception: " << e.what() << std::endl;
                }
            }
            
            
            
        }
    }
};

class NetworkConnectionPool {
private:
    int server_fd_;
    int epoll_fd_;
    std::thread epoll_thread_;
    std::atomic<bool> running_{false};

    std::unique_ptr<WorkStealingThreadPool> worker_pool_;
    std::unordered_map<int, std::chrono::steady_clock::time_point> connections_;
    std::mutex connections_mutex_;

    int port_;
    size_t max_events_;
    std::chrono::seconds connection_timeout_;
    // 性能统计
    std::atomic<uint64_t> total_connections_{0};
    std::atomic<uint64_t> total_requests_{0};
    std::atomic<uint64_t> total_bytes_received_{0};
  
public:
    NetworkConnectionPool(int port, size_t num_workers = std::thread::hardware_concurrency())
     : port_(port), max_events(1024), connection_timeout_(std::chrono::seconds(30)){
        worker_pool_ = std::make_unique<WorkStealingThreadPool>(
            num_events,
            [this](std::unique_ptr<NetWordRequest> req){
                handleRequest(std::move(req));
            }
        );
        initServer();
    } 
    void start() {
        running_.store(true, std::memory_order_relaxed);
        thread_pool_->start();
        epoll_thread_ = std::thread(&NetworkConnectionPool::epollLoop, this);
    }

    void stop()
    {
        running_.store(false, std::memory_order_relaxed);
        if (epoll_thread_.joinable()) epoll_thread_.join();
        thread_pool_->stop();
        cleanup();
    }

private:
    void initServer() {
        server_fd_ = socket(AF_INET, SOCK_STREAM, 0);
        if (server_fd_ == -1) {
            throw std::runtime_error("Failed to create server socket");
        }
        int pot = 1;
        setscokopt(server_fd_, SOL_SOCKET, SO_REUSEADDR, &pot, sizeof(pot));
        setNonBlocking(server_fd_);

        sockaddr_in addr;
        addr.sin_family = AF_INET;
        addr.sin_addr.s_addr = INADDR_ANY;
        addr.sin_port = htons(port_);

        if (bind(server_fd_, (sockaddr*)&addr, sizeof(addr)))
        {
            throw std::runtime_error("Failed to bind server socket");
        }
        if (listen(server_fd_, SOMAXCONN) == -1) 
        {
            throw std::runtime_error("Failed to listen on server socket");
        }
        epoll_fd_ = epoll_create1(EPOLL_CLOEXEC);
        if (epoll_fd_ == -1) {
            throw std::runtime_error("Failed to create epoll instance");
        }
        epoll_event event;
        event.events = EPOLLIN | EPOLLET;
        event.adta.fd = server_fd_;
        epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, server_fd_, &event);

    }

    void epollLoop() {
        std::vector<epoll_event> events(max_events_);
        while (running_) {
            int num_events = epoll_wait(epoll_fd_, events.data(), max_events_, -1);
            for (int i = 0; i < num_events; ++i){
                int fd = events[i].data.fd;
                uint32_t mask = events[i].events;
                if (fd == server_fd_) {
                    handleNewConnection();
                } else {
                    if (mask & EPOLLIN) {
                        handleClientData(fd);
                    }
                    if (mask & (EPOLLHUP | EPOLLRDHUP)) {
                        closeConnection(fd);
                    }
                }
            }        
        }
        cleanupTimeoutConnections();
    }

    void handleNewConnection() {
        while(true) {
            sockaddr_in client_addr;
            socklen_t client_len = sizeof(client_addr);
            int client_fd = accept(server_fd_, (sockaddr*)&client_addr, &client_len);
            setNonBlocking(client_fd);
            epoll_event event;
            event.events = EPOLLIN | EPOLLET;
            event.data.fd = client_fd;
            epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, client_fd, &event);
            // 记录连接
            {
                std::lock_guard<std::mutex> lock(connections_mutex_);
                connections_[client_fd] = std::chrono::steady_clock::now();
            }
            
            total_connections_.fetch_add(1);
        }
    }

    void handleClientData(int fd) {
        char buffer[4096];
        std::string data;

        while(true) {
            ssize_t bytes_read = read(fd, buffer, sizeof(buffer));
            if (bytes_read > 0) {
                data.append(buffer, bytes_read);
                total_bytes_received_.fetch_add(bytes_read);
            } else if (bytes_read == 0) {
                closeConnection(fd);
                break;
            } else {
                closeConnection(fd);
                break;
            }
        }
        if (!data.empty()) {
            auto req = std::make_unique<NetWordRequest>(fd, std::move(data), std::chrono::steady_clock::now() + connection_timeout_);
            thread_pool_->submit(std::move(req));
            
        }

    }

    virtual void handleRequest(std::unique_ptr<NetWordRequest> req) {
        throw std::runtime_error("Not implemented");
    }

};

}

namespace raft {

struct VoteRequest {
    int nodeId;
    int term;
    int lastLogIndex = 0;  // Raft协议需要
    int lastLogTerm = 0;   // Raft协议需要
};

struct VoteResponse {
    int nodeId;
    int term;
    bool voteGranted;
};

struct HeartbeatRequest {
    int nodeId;
    int term;
    int prevLogIndex = 0;  // 用于日志复制
    int prevLogTerm = 0;   // 用于日志复制
    int leaderCommit = 0;  // leader的commitIndex
};

struct HeartbeatResponse {
    int nodeId;
    int term;
    bool success;  // 修复：添加success字段
};

enum Role {
    FOLLOWER,
    CANDIDATE,
    LEADER
};

enum State {
    RUNNING,
    STOPPED
}; // 修复：添加分号

class Raft {
public:
    Raft(int nodeId, std::vector<int> nodes) : 
        nodeId_(nodeId), 
        nodes_(nodes),
        currentTerm_(0),
        votedFor_(-1),
        role_(FOLLOWER),
        leaderId_(-1),
        state_(STOPPED),
        maxElectionTimeout_(2000),
        minElectionTimeout_(1000) {
        resetElectionTimeout();
    }

    ~Raft() {
        stop();
    }

    void start() {
        {
            std::lock_guard<std::mutex> lock(mutex_raft_info_);
            if (state_ == RUNNING) return;
            state_ = RUNNING;
            resetElectionTimeout();
        }
        
        // 启动选举超时线程
        electionTimeoutThread_ = std::thread(&Raft::electionTimeoutLoop, this);
        
        // 如果是leader，启动心跳线程
        if (role_ == LEADER) {
            startHeartbeatThread();
        }
    }

    void stop() {
        {
            std::lock_guard<std::mutex> lock(mutex_raft_info_);
            state_ = STOPPED;
        }
        
        if (electionTimeoutThread_.joinable()) {
            electionTimeoutThread_.join();
        }
        
        if (heartbeatThread_.joinable()) {
            heartbeatThread_.join();
        }
    }

    // 修复：正确的选举超时检查
    bool isElectionTimeoutNeedLocked() {
        auto now = std::chrono::steady_clock::now();
        auto elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(
            now - electionTimeoutStart_).count();
        return elapsed >= electionTimeout_;
    }

    void resetElectionTimeoutNeedLocked() {
        electionTimeoutStart_ = std::chrono::steady_clock::now();
        electionTimeout_ = getRandomElectionTimeoutNeedLocked();
    }

    void becomeFollowerNeedLocked(int term) {
        if (term < currentTerm_) {
            return;
        }
        
        bool wasLeader = (role_ == LEADER);
        role_ = FOLLOWER;
        leaderId_ = -1;
        
        if (term > currentTerm_) {
            votedFor_ = -1;
            currentTerm_ = term;
        }
        
        resetElectionTimeoutNeedLocked();
        
        // 如果之前是leader，需要停止心跳
        if (wasLeader && heartbeatThread_.joinable()) {
            // 心跳线程会自己检测到角色变化并退出
        }
    }

    void becomeCandidateNeedLocked() {
        if (role_ == LEADER) {
            return;
        }
        currentTerm_++;
        votedFor_ = nodeId_;
        role_ = CANDIDATE;
        resetElectionTimeoutNeedLocked();
        
        std::cout << "Node " << nodeId_ << " becomes candidate for term " << currentTerm_ << std::endl;
    }

    void becomeLeaderNeedLocked() {
        if (role_ != CANDIDATE) {
            return;
        }
        role_ = LEADER;
        leaderId_ = nodeId_;
        resetElectionTimeoutNeedLocked();
        
        std::cout << "Node " << nodeId_ << " becomes leader for term " << currentTerm_ << std::endl;
        
        // 启动心跳线程
        startHeartbeatThread();
    }

    int getRandomElectionTimeoutNeedLocked() {
        static std::random_device rd;
        static std::mt19937 gen(rd());
        std::uniform_int_distribution<> dis(minElectionTimeout_, maxElectionTimeout_);
        return dis(gen);
    }

    // 修复：简化选举逻辑
    void startElection() {
        std::lock_guard<std::mutex> lock(mutex_raft_info_);
        
        becomeCandidateNeedLocked();
        int electionTerm = currentTerm_;
        
        // 为自己投票
        std::atomic<int> voteCount{1};
        
        std::cout << "Node " << nodeId_ << " starting election for term " << electionTerm << std::endl;
        
        // 向其他节点请求投票
        for (int nodeId : nodes_) {
            if (nodeId == nodeId_) continue;
            
            VoteRequest request;
            request.nodeId = nodeId_;
            request.term = electionTerm;
            request.lastLogIndex = 0; // TODO: 实现日志时填入真实值
            request.lastLogTerm = 0;  // TODO: 实现日志时填入真实值
            
            // 异步发送投票请求
            std::thread([this, nodeId, request, &voteCount, electionTerm]() {
                VoteResponse response = sendVoteRequestSync(nodeId, request);
                
                std::lock_guard<std::mutex> lock(mutex_raft_info_);
                
                // 检查状态是否仍然有效
                if (role_ != CANDIDATE || currentTerm_ != electionTerm) {
                    return;
                }
                
                if (response.term > currentTerm_) {
                    becomeFollowerNeedLocked(response.term);
                    return;
                }
                
                if (response.voteGranted) {
                    int newVoteCount = voteCount.fetch_add(1) + 1;
                    std::cout << "Node " << nodeId_ << " received vote from " << nodeId 
                              << ", total votes: " << newVoteCount << std::endl;
                    
                    if (newVoteCount > nodes_.size() / 2) {
                        becomeLeaderNeedLocked();
                    }
                }
            }).detach();
        }
    }

    // 处理投票请求
    VoteResponse receiveVoteRequest(const VoteRequest& voteRequest) {
        std::lock_guard<std::mutex> lock(mutex_raft_info_);
        
        VoteResponse response;
        response.nodeId = nodeId_;
        response.term = currentTerm_;
        response.voteGranted = false;
        
        std::cout << "Node " << nodeId_ << " received vote request from " 
                  << voteRequest.nodeId << " for term " << voteRequest.term << std::endl;
        
        // 如果请求的term小于当前term，拒绝投票
        if (voteRequest.term < currentTerm_) {
            std::cout << "Node " << nodeId_ << " rejects vote: term too old" << std::endl;
            return response;
        }
        
        // 如果请求的term大于当前term，更新状态
        if (voteRequest.term > currentTerm_) {
            becomeFollowerNeedLocked(voteRequest.term);
        }
        
        response.term = currentTerm_;
        
        // 投票条件：还没投票 或 已经投给了这个候选人
        if (votedFor_ == -1 || votedFor_ == voteRequest.nodeId) {
            votedFor_ = voteRequest.nodeId;
            response.voteGranted = true;
            resetElectionTimeoutNeedLocked();
            std::cout << "Node " << nodeId_ << " grants vote to " << voteRequest.nodeId << std::endl;
        } else {
            std::cout << "Node " << nodeId_ << " rejects vote: already voted for " << votedFor_ << std::endl;
        }
        
        return response;
    }

    // 修复：心跳逻辑
    void sendHeartbeats() {
        std::lock_guard<std::mutex> lock(mutex_raft_info_);
        
        if (role_ != LEADER) {
            return;
        }
        
        int currentHeartbeatTerm = currentTerm_;
        
        for (int nodeId : nodes_) {
            if (nodeId == nodeId_) continue;
            
            HeartbeatRequest request;
            request.nodeId = nodeId_;
            request.term = currentHeartbeatTerm;
            request.prevLogIndex = 0;  // TODO: 实现日志时填入
            request.prevLogTerm = 0;   // TODO: 实现日志时填入
            request.leaderCommit = 0;  // TODO: 实现日志时填入
            
            // 异步发送心跳
            std::thread([this, nodeId, request, currentHeartbeatTerm]() {
                HeartbeatResponse response = sendHeartbeatRequestSync(nodeId, request);
                
                std::lock_guard<std::mutex> lock(mutex_raft_info_);
                
                if (role_ != LEADER || currentTerm_ != currentHeartbeatTerm) {
                    return;
                }
                
                if (response.term > currentTerm_) {
                    becomeFollowerNeedLocked(response.term);
                }
            }).detach();
        }
    }

    // 处理心跳请求
    HeartbeatResponse receiveHeartbeatRequest(const HeartbeatRequest& request) {
        std::lock_guard<std::mutex> lock(mutex_raft_info_);
        
        HeartbeatResponse response;
        response.nodeId = nodeId_;
        response.term = currentTerm_;
        response.success = false;
        
        // 如果心跳的term大于当前term，更新状态
        if (request.term > currentTerm_) {
            becomeFollowerNeedLocked(request.term);
        }
        
        response.term = currentTerm_;
        
        // 如果term匹配，接受心跳
        if (request.term == currentTerm_) {
            // 重置选举超时
            resetElectionTimeoutNeedLocked();
            leaderId_ = request.nodeId;
            response.success = true;
            
            // 如果是候选人，变成跟随者
            if (role_ == CANDIDATE) {
                role_ = FOLLOWER;
            }
        }
        
        return response;
    }

private:
    void electionTimeoutLoop() {
        while (true) {
            {
                std::lock_guard<std::mutex> lock(mutex_raft_info_);
                if (state_ == STOPPED) break;
                
                // 只有非Leader需要检查选举超时
                if (role_ != LEADER && isElectionTimeoutNeedLocked()) {
                    // 释放锁后开始选举（避免在锁内启动新线程）
                }
            }
            
            // 如果需要选举，在锁外启动
            {
                std::lock_guard<std::mutex> lock(mutex_raft_info_);
                if (state_ == STOPPED) break;
                if (role_ != LEADER && isElectionTimeoutNeedLocked()) {
                    startElection();
                }
            }
            
            std::this_thread::sleep_for(std::chrono::milliseconds(50));
        }
    }

    void startHeartbeatThread() {
        if (heartbeatThread_.joinable()) {
            return; // 已经在运行
        }
        
        heartbeatThread_ = std::thread(&Raft::heartbeatLoop, this);
    }

    void heartbeatLoop() {
        while (true) {
            {
                std::lock_guard<std::mutex> lock(mutex_raft_info_);
                if (state_ == STOPPED || role_ != LEADER) {
                    break;
                }
            }
            
            sendHeartbeats();
            std::this_thread::sleep_for(std::chrono::milliseconds(150)); // 心跳间隔
        }
    }

    // 模拟网络调用（实际应该通过网络层）
    VoteResponse sendVoteRequestSync(int targetNodeId, const VoteRequest& request) {
        // 这里应该通过网络发送请求
        // 暂时返回模拟响应
        VoteResponse response;
        response.nodeId = targetNodeId;
        response.term = currentTerm_;
        response.voteGranted = (rand() % 3 == 0); // 33%概率投票，仅用于测试
        return response;
    }

    HeartbeatResponse sendHeartbeatRequestSync(int targetNodeId, const HeartbeatRequest& request) {
        // 这里应该通过网络发送请求
        HeartbeatResponse response;
        response.nodeId = targetNodeId;
        response.term = currentTerm_;
        response.success = true;
        return response;
    }

    void resetElectionTimeout() {
        resetElectionTimeoutNeedLocked();
    }

public:
    // 成员变量
    std::mutex mutex_raft_info_;
    
    // 节点信息
    std::vector<int> nodes_;
    State state_;
    int nodeId_;
    
    // Raft状态
    int currentTerm_;
    int votedFor_;
    Role role_;
    int leaderId_;
    
    // 选举超时
    std::chrono::steady_clock::time_point electionTimeoutStart_;
    int electionTimeout_;
    int maxElectionTimeout_;
    int minElectionTimeout_;
    
    // 线程
    std::thread electionTimeoutThread_;
    std::thread heartbeatThread_;
};

} // namespace raft
